package net.bwie.gongdan.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
/*
    大数据-用户画像-12-达摩盘家庭特征
 */

public class ItemBabyLabelDimJob {
    public static void main(String[] args) {
        // 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 输入表  input 映射到kafka消息队列
        createInputTable(tabEnv);
        // 数据处理
        Table reportTable = handle(tabEnv);
        // 输出表  output  映射到clickhouse表
        createOutputTable(tabEnv);
        // 保存数据
        saveToDoris(tabEnv, reportTable);
    }

    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
        tabEnv.createTemporaryView("report_table", reportTable);
        tabEnv.executeSql("INSERT INTO item_attributes SELECT * FROM report_table");
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE item_attributes (\n" +
                "    `item_id`       STRING COMMENT '商品ID',\n" +
                "    `item_name`     STRING COMMENT '商品名称',\n" +
                "    `category_id`   STRING COMMENT '类目ID',\n" +
                "    `category_path` STRING COMMENT '完整类目路径',\n" +
                "    `keywords`      STRING COMMENT '商品关键词',\n" +
                "    `description`   STRING COMMENT '商品描述',\n" +
                "    `price`         DECIMAL(10, 2) COMMENT '商品价格',\n" +
                "    `is_maternal`   BOOLEAN COMMENT '是否母婴类商品',\n" +
                "    PRIMARY KEY (item_id) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'item_attributes',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");
    }

    private static Table handle(TableEnvironment tabEnv) {
//        Table table = tabEnv.sqlQuery("select * from user_info");
//        tabEnv.createTemporaryView("user_info", table);

        Table table1 = tabEnv.sqlQuery("select * from item_attributes_kafka where category_path like '%母婴用品%'");

        return table1;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE item_attributes_kafka\n" +
                "(\n" +
                "    item_id       STRING COMMENT '商品ID',\n" +
                "    item_name     STRING COMMENT '商品名称',\n" +
                "    category_id   STRING COMMENT '类目ID',\n" +
                "    category_path STRING COMMENT '完整类目路径',\n" +
                "    keywords      STRING COMMENT '商品关键词',\n" +
                "    description   STRING COMMENT '商品描述',\n" +
                "    price         DECIMAL(10, 2) COMMENT '商品价格',\n" +
                "    is_maternal   BOOLEAN COMMENT '是否母婴类商品'\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'item_attributes_topic',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                "    'properties.group.id' = 'item-attributes-consumer',\n" +
                "    'format' = 'json',  -- 或其他格式如'avro','protobuf'\n" +
                "    'scan.startup.mode' = 'earliest-offset',  -- 从最早偏移量开始消费\n" +
                "    'json.ignore-parse-errors' = 'true',  -- 可选:忽略解析错误\n" +
                "    'json.timestamp-format.standard' = 'ISO-8601'  -- 时间格式\n" +
                ")");
    }

    private static TableEnvironment getTableEnv() {
        // 环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        // 时区设置
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并发设置
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态保存时间
        configuration.setString("table.exec.state.ttl", "5s");
        // 返回值
        return tabEnv;
    }
}
